package com.cap.bts.crmdp.webservice.dc;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.cap.bts.crmdp.dealer.mars.domain.DataConversionLog;
import com.cap.bts.crmdp.dealer.mars.service.IDataConversionLogService;
import com.cap.bts.crmdp.webservice.dto.DtCrmDto;
import com.cap.bts.framework.codemapping.service.ICodeMapping;
import com.cap.bts.framework.common.util.ApplicationContextUtil;

public class DataConversionQueue {
	private static Logger logger = Logger.getLogger(DataConversionQueue.class);

	private static final LinkedBlockingQueue<DtCrmDto> PROCESS_DATA_QUEUE = new LinkedBlockingQueue<DtCrmDto>();
	
	protected static final Map<String, DataConversionLog> BATCH_LOG_DATA = new HashMap<String, DataConversionLog>();
	protected static final List<String> PROCESS_BATCH = new ArrayList<String>();

	private static boolean beginFlag = false;
	
	public static void insertData(DataConversionLog dataConversionLog){
		try{
			IDataConversionLogService dataConversionLogServiceImpl = (IDataConversionLogService) ApplicationContextUtil.getBean("dataConversionLogServiceImpl");
			//相同批次号的不重复处理
			if(!PROCESS_BATCH.contains(dataConversionLog.getBatchId())){
				DtCrmDto dto = JSON.parseObject(dataConversionLog.getParamText(), DtCrmDto.class);
				
				dataConversionLog.setStatus("2");
				
				dataConversionLogServiceImpl.update(dataConversionLog);
				
				BATCH_LOG_DATA.put(dto.getBatchId(), dataConversionLog);
				
				PROCESS_DATA_QUEUE.add(dto);
				PROCESS_BATCH.add(dataConversionLog.getBatchId());
			}else{
				dataConversionLog.setStatus("5");//已有相同批次号处理过
				
				dataConversionLogServiceImpl.update(dataConversionLog);
			}
			
		}catch (Exception e) {
			logger.error("", e);
		}
	}
	
	public static void start(){
		if(!beginFlag){
			beginFlag = true;
			new Thread(){
				@Override
				public void run(){
					logger.info("start data conversion thread.");
					new DataConversionQueue().process();
				}
			}.start();
			
			new DataConversionLoadDataThread().start();
			
		}
	}
	
	public static DataConversionLog getBatchInitDate(String batchId){
		return BATCH_LOG_DATA.get(batchId);
	}
	
	private void process() {
		while (true) {
			try {
				// 队列自动阻塞，直到获取到数据
				DtCrmDto dto = PROCESS_DATA_QUEUE.take();
				logger.info("start process:" + dto.getBatchId());
				new DataConversionProcess(dto).process();
				
			} catch (Exception e) {
				logger.error("", e);
			}
		}
	}
}
